package reactor.mytest;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

public class ReactorFluxSubscribe {
	
	public static void main(String[] args) {
		ExecutorService service = Executors.newFixedThreadPool(5);
		Scheduler scheduler = Schedulers.fromExecutor(service);
		Flux.create(sink -> {
			for(int i=0;i<100000;i++){
				 sink.next(Thread.currentThread().getName());
			} 
		    sink.complete();
		})
		.publishOn(scheduler)
		.flatMap(x -> {   return  Flux.just(String.format("[%s] %s", Thread.currentThread().getName(), x));})
		.publishOn(Schedulers.elastic())
		.flatMap(x ->{   return  Flux.just(String.format("[%s] %s", Thread.currentThread().getName(), x));})
		.subscribeOn(scheduler)
		.toStream()
		.forEach(System.out::println);
		service.shutdown();
	}
}
